package ex.tableapi.window;

import ex.tableapi.ApiFrame;
import org.apache.flink.table.api.Session;
import org.apache.flink.table.api.Table;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;

public class ExSessionWindows extends ApiFrame {
    private String outName = "printOutTable";

    public static void main(String[] args) {
        ExSessionWindows demo = new ExSessionWindows();
        demo.getEnv();
        Table orders = demo.registerMysql("myorder", "orders");

        Table result = demo.exSessionWindows(orders);

        demo.registerConsole(demo.createPrintOutDDL(result.getResolvedSchema().toString()), demo.outName);
        result.executeInsert(demo.outName);
    }


    /**
     * 会话窗口没有固定的大小，但它们的边界由不活动的时间间隔定义，
     * 即，如果在定义的间隔期内没有出现任何事件，则会话窗口将关闭。
     * 例如，当在 30 分钟不活动后观察到某行时，间隔为 30 分钟的会话窗口将启动（否则该行将添加到现有窗口中），
     * 如果在 30 分钟内未添加任何行，则关闭该窗口。会话窗口可以在事件时间或处理时间上工作。
     *
     * @param orders
     * @return
     */
    private Table exSessionWindows(Table orders) {
        //对应registerMysql 水印字段调整为 WATERMARK FOR proctime AS proctime - INTERVAL '5' SECOND"
//        Table result = orders.window(Session.withGap(lit(10).minutes()).on($("proctime")).as("w"))
        Table result = orders.window(Session.withGap(lit(10).minutes()).on($("rowtime")).as("w"))
                .groupBy($("a"), $("w")) // group by key and window
                .select(
                        $("a"),
                        $("w").start(),
                        $("w").end(),
                        $("w").rowtime(),
                        $("b").count()
                );
// Session Event-time Window

/**
 //proctime
 1> +I[zy, 2024-01-26T18:46:33, 2024-01-26T18:56:33, 2024-01-26T18:56:32.999, 6]
 1> +I[liji, 2024-01-26T18:46:33, 2024-01-26T18:56:54, 2024-01-26T18:56:53.999, 6]
 // rowtime
 1> +I[liji, 2024-01-24T19:04:19, 2024-01-24T19:14:24, 2024-01-24T19:14:23.999, 2]
 1> +I[liji, 2024-01-24T19:35:51, 2024-01-24T19:45:51, 2024-01-24T19:45:50.999, 1]
 1> +I[liji, 2024-01-25T19:04:19, 2024-01-25T19:14:29, 2024-01-25T19:14:28.999, 2]
 1> +I[zy, 2024-01-25T19:03:55, 2024-01-25T19:15:08, 2024-01-25T19:15:07.999, 2]
 1> +I[zy, 2024-01-25T19:35:15, 2024-01-25T19:45:15, 2024-01-25T19:45:14.999, 1]
 1> +I[zy, 2024-01-25T21:05:26, 2024-01-25T21:15:26, 2024-01-25T21:15:25.999, 1]
 1> +I[zy, 2024-01-25T21:35:36, 2024-01-25T21:45:36, 2024-01-25T21:45:35.999, 1]
 1> +I[zy, 2024-01-26T19:05:07, 2024-01-26T19:15:07, 2024-01-26T19:15:06.999, 1]
 */
        return result;
    }

}
